package com.s666.demo.service.impl;

import com.s666.demo.service.RocketMQService;
import lombok.extern.slf4j.Slf4j;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;

import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;


/**
 * @author lijin
 */
@Service("RocketMQService")
@Slf4j
public class RocketMQServiceImpl implements RocketMQService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    //////////////// 普通消息发送 start
    public void product(String topic, String msg) {
        // 发送普通消息: 异步
        rocketMQTemplate.convertAndSend(topic,msg);
        System.out.println("Message sent: " + msg);

        // 同步
        //SendResult result = rocketMQTemplate.syncSend("TopicTest",  "Hello RocketMQ");
    }
    //////////////// 普通消息发送 end


    //////////////// 事务消息发送 start
    // 它会自动调用 executeLocalTransaction 方法
    public void createOrderWithTransaction(String topic, String msg) {
        Message<String> message = MessageBuilder.withPayload("OrderCreated:" + msg).build();
        // 发送事务消息
        rocketMQTemplate.sendMessageInTransaction("TopicTest", message, null);
    }

    @RocketMQTransactionListener
    public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                // 执行本地事务（例如扣减库存）
                System.out.println("Executing local transaction for message: " + new String((byte[]) msg.getPayload()));
                // 假设扣减库存成功
                return RocketMQLocalTransactionState.COMMIT;
            } catch (Exception e) {
                // 如果扣减库存失败，回滚消息
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // 检查本地事务状态（例如检查库存是否扣减成功）
            System.out.println("Checking local transaction for message: " + new String((byte[]) msg.getPayload()));
            return RocketMQLocalTransactionState.COMMIT;
        }
    }
    //////////////// 事务消息发送 end

}
@Service
/* 这里打开就是消费者
@RocketMQMessageListener(
        topic = "TopicTest",
        consumerGroup = "my-consumer-group"
        selectorType = SelectorType.TAG, // 按标签过滤
        selectorExpression = "TagA || TagB"
)
*/
class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received  message: " + message);
    }
}

